package streaming.day01

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import util.Jpools

/** 带历史状态统计的一个优化 + redis
  * sheep.Old @ 64341393
  * Created 2018/4/21
  */
object WordCountStateRedis {

    def main(args: Array[String]): Unit = {

        // master 需要 2 个核, 以防止饥饿情况（starvation scenario）.
        // 1一个核用于运行当前程序，一个核用于接收数据
        val conf = new SparkConf()
          .setAppName("wordcountshow")
          .setMaster("local[*]")

        // 第二个参数为批次时间间隔，多长时间的数据集为一个批次，这个时间不能拍脑门设置，科学合理的设置,一般会比较短
        val ssc = new StreamingContext(conf, Seconds(2)) // 2秒钟一个批次

        // 接收受数据
        val stream: ReceiverInputDStream[String] = ssc.socketTextStream("10.172.50.12", 44444)

        stream.foreachRDD(rdd => {
            val result = rdd.flatMap(line => line.split(" ")).map((_, 1)).reduceByKey(_ + _)
            // 把计算好的当前批次结果写入到redis中
            result.foreachPartition(iter => {
                // 每个分区从池子里面获取一个连接对象
                val jedis = Jpools.getJedis
                iter.foreach(tp => jedis.hincrBy("wordcount", tp._1, tp._2))
                // 用完之后还回连接
                jedis.close()
            })
        })


        // 启动程序
        ssc.start() // 启动当前main方法这个主程序
        // 不要死
        ssc.awaitTermination() // 主程序阻塞
    }

}
